/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "fastdds/dds/core/StackAllocatedSequence.hpp"
#include <fastdds/dds/subscriber/DataReader.hpp>
#include "dal_fastdds/listener_impl/DataReaderListenerImplDbhInfo.hpp"
#include "dal_fastdds/utils/log.hpp"
#include "dal_dmabufheap/DbhNotifyMsgDef.hpp"
#include "dal_dmabufheap/FdStrConverter.hpp"

DataReaderListenerImplDbhInfo::DataReaderListenerImplDbhInfo(
    const message_type_support_callbacks_t* ts_callbacks, DalDbhReader* dbh_reader)
    : dbh_info_ts_callbacks_(ts_callbacks), dbh_reader_(dbh_reader)
{
    const char* env_var_value = std::getenv("DAL_DBH_DEBUG");
    if (env_var_value != nullptr && std::atoi(env_var_value) == 1) {
        is_debug_print_enabled_ = true;
    }
}

DataReaderListenerImplDbhInfo::~DataReaderListenerImplDbhInfo() {}

void DataReaderListenerImplDbhInfo::on_data_available(DataReader* reader)
{
    SampleInfoSeq info_seq{1};
    dal_fastrtps_shared_cpp::SerializedData sdata;
    sdata.is_cdr_buffer = false;
    sdata.data = &dbh_info_msg_;
    sdata.impl = dbh_info_ts_callbacks_;
    eprosima::fastdds::dds::StackAllocatedSequence<void*, 1> data_values;
    const_cast<void**>(data_values.buffer())[0] = &sdata;

    ReturnCode_t ret;
    while (true) {
        ret = reader->take(data_values, info_seq, 1);
        if (ret == ReturnCode_t::RETCODE_NO_DATA) {
            break;
        }

        if (ret != ReturnCode_t::RETCODE_OK) {
            LOG_ERROR_WITH_LOCATION()
                << "Error reading data from datareader ret = " << ret() << std::endl;
            return;
        }

        if (!dbh_reader_) {
            LOG_ERROR_WITH_LOCATION() << "dbh_reader is nullptr" << std::endl;
            return;
        }

        if (dbh_info_msg_.event == static_cast<uint32_t>(DbhNotifyMsgDef::RESIZE_INFO)) {
            int32_t global_fd = std::stoi(dbh_info_msg_.data);
            if (global_fd != -1) {
                if (is_debug_print_enabled_) {
                    std::cout << "[dbh info reader <------] event: RESIZE_INFO, fd = " << global_fd
                              << std::endl;
                }

                dbh_reader_->DeleteDbhBlockFromMap(global_fd);
            }
        } else if (dbh_info_msg_.event == static_cast<uint32_t>(DbhNotifyMsgDef::DELETE_FDS)) {
            std::vector<int32_t> fds;
            if (is_debug_print_enabled_) {
                std::cout << "[dbh info reader <------] event: DELETE_FDS, fds = "
                          << dbh_info_msg_.data << std::endl;
            }

            if (Str2Fds(dbh_info_msg_.data, fds)) {
                for (auto fd : fds) {
                    dbh_reader_->DeleteDbhBlockFromMap(fd);
                }
            } else {
                LOG_ERROR_WITH_LOCATION() << "Error converting string to fds" << std::endl;
            }
        }

        continue;
    }
}